package com.shujia.sql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object Demo04Connectors {
  def main(args: Array[String]): Unit = {
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() // 使用Blink的SQL解析器
      .inStreamingMode() // 开启流处理模式
      .build()
    // 构建Table Env
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    // 读文件
    bsTableEnv
      .executeSql(
        """
          |-- 有界流
          |CREATE TABLE stu_json_table (
          |  id INT,
          |  name STRING,
          |  age STRING,
          |  gender STRING,
          |  clazz STRING
          |)
          |WITH (
          |  'connector' = 'filesystem',
          |  'path' = 'Flink/data/stu/students.json',
          |  'format' = 'json'
          |)
          |""".stripMargin)


    // 读MySQL
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE stu_mysql_table (
          |  id INT,
          |  name STRING,
          |  age INT,
          |  gender STRING,
          |  clazz STRING
          |) WITH (
          |   'connector' = 'jdbc',
          |   'url' = 'jdbc:mysql://master:3306/student',
          |   'table-name' = 'student',
          |   'username' = 'root',
          |   'password' = '123456'
          |)
          |""".stripMargin)

    // 使用DataGen随机生成数据
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE datagen (
          | f_sequence INT,
          | f_random INT,
          | f_random_str STRING
          |) WITH (
          | 'connector' = 'datagen',
          | 'rows-per-second'='5',
          |
          | 'fields.f_sequence.kind'='sequence',
          | 'fields.f_sequence.start'='1',
          | 'fields.f_sequence.end'='1000',
          |
          | 'fields.f_random.min'='1',
          | 'fields.f_random.max'='1000',
          |
          | 'fields.f_random_str.length'='10'
          |)
          |""".stripMargin)

    // 读取HBase的数据
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE stu_hbase_table (
          | id INT,
          | info ROW<name String,age String,gender String,clazz String>,
          | PRIMARY KEY (id) NOT ENFORCED
          |) WITH (
          | 'connector' = 'hbase-1.4',
          | 'table-name' = 'stu',
          | 'zookeeper.quorum' = 'master:2181,node1:2181,node2:2181'
          |)
          |""".stripMargin)

    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE print_table1 WITH ('connector' = 'print')
          |LIKE stu_json_table (EXCLUDING ALL)
          |""".stripMargin)


    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE print_table2 WITH ('connector' = 'print')
          |LIKE stu_mysql_table (EXCLUDING ALL)
          |""".stripMargin)

    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE print_table3 WITH ('connector' = 'print')
          |LIKE datagen (EXCLUDING ALL)
          |""".stripMargin)

    // 将数据写入blackhole 主要用于测试性能
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE blackhole_table WITH ('connector' = 'blackhole')
          |LIKE datagen (EXCLUDING ALL)
          |""".stripMargin)

    // 打印HBase的数据 需要手动指定字段格式
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE hbase_print_table(
          |  clazz STRING,
          |  cnt BIGINT
          |)
          |WITH ('connector' = 'print')
          |""".stripMargin)

    // 将数据写入HBase  create 'stu_clazz_cnt','cf1'
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE clazz_cnt_hbase_sink_table (
          | clazz STRING,
          | cf1 ROW<clazz2 String,cnt BIGINT>,
          | PRIMARY KEY (clazz) NOT ENFORCED
          |) WITH (
          | 'connector' = 'hbase-1.4',
          | 'table-name' = 'stu_clazz_cnt',
          | 'zookeeper.quorum' = 'master:2181,node1:2181,node2:2181'
          |)
          |""".stripMargin)

    //    bsTableEnv
    //      .executeSql(
    //        """
    //          |insert into print_table1
    //          |select * from stu_json_table
    //          |""".stripMargin)


    //    bsTableEnv
    //      .executeSql(
    //        """
    //          |insert into print_table2
    //          |select * from stu_mysql_table
    //          |""".stripMargin)

    //    bsTableEnv
    //      .executeSql(
    //        """
    //          |insert into print_table3
    //          |select * from datagen
    //          |""".stripMargin)

    //    bsTableEnv
    //      .executeSql(
    //        """
    //          |insert into blackhole_table
    //          |select * from datagen
    //          |""".stripMargin)

    //    bsTableEnv
    //      .executeSql(
    //        """
    //          |insert into hbase_print_table
    //          |select clazz,count(1) as cnt from stu_hbase_table
    //          |group by clazz
    //          |""".stripMargin)

    // 读取json格式的学生数据 并统计班级人数 最终写入HBase
    bsTableEnv
      .executeSql(
        """
          |insert into clazz_cnt_hbase_sink_table
          |select clazz
          |       ,Row(clazz2,cnt)
          |from (
          |select clazz,clazz as clazz2,count(id) as cnt from stu_json_table group by clazz
          |) t1
          |""".stripMargin)

  }

}
